﻿using Confluent.Kafka;
using Leo.Common.RabbitMQ;
using Microsoft.AspNetCore.Mvc;
using RabbitMQ.Client;
using System.Text;
using System.Threading.Tasks;

namespace JwtAuth.Controllers
{

    /// <summary>
    /// 生产者 端 Demo
    /// </summary>
    [Route("api/product")]
    [ApiController]
    public class ProducerController : ControllerBase
    {
        [HttpGet("send/message")]
        public async Task<IActionResult> GetConsumerTest(string messages)
        {
            //创建RabbitMQ  链接            
            RabbitConnection conn = new RabbitConnection(new RabbitConfig(@"host=127.0.0.1;port=5672;user=admin;pwd=123456;vhost=myhost;qos=1"));
            var channel = await conn.GetChannelAsync();

            string queueName = "queueNameTest";
            string exchangeName = "topic_exchange";
            string routingKey = "routing_key.*";

            //声明队列
            await channel.QueueDeclareAsync(queueName, false, false, false, null);

            //声明交换机
            await channel.ExchangeDeclareAsync(exchangeName, "topic", false, false, null);

            // 将队列和交换机绑定
            await channel.QueueBindAsync(queueName, exchangeName, routingKey);


            Console.WriteLine($@"RabbitMQ 连接成功；Exchange:{exchangeName};Queue:{queueName},RoutingKey:{routingKey}");


            // 向 交换机 发消息
            var sendByte = Encoding.UTF8.GetBytes(messages);
            BasicProperties basicProperties = new BasicProperties { };
            await channel.BasicPublishAsync(exchangeName, "routing_key.one", false, basicProperties, sendByte);


            return Ok("message sen ok");
        }


        /// <summary>
        /// 发消息 到kafka
        /// </summary>
        /// <param name="content"></param>
        /// <returns></returns>
        [HttpGet("send/kafka")]
        public async Task<IActionResult> SendKafkaMessage(string content)
        {
            string brokerList = "localhost:9092";
            string topicName = "logkafka";

            //生产者配置
            var config = new ProducerConfig
            {
                //kafka节点
                BootstrapServers = brokerList,
                //ack机制，0不等服务器确认，1主节点确认返回ack,-1全部节点同步完返回ack
                Acks = Acks.None
            };


            using (var producer = new ProducerBuilder<string, string>(config).Build())
            {
                try
                {
                    //key要给值，根据key做负载均衡，不然如果多节点，key不给值会全部写有一个分区
                    var deliveryReport = await producer.ProduceAsync(topicName, new Message<string, string> { Key = (new Random().Next(1, 10)).ToString(), Value = content });
                    Console.WriteLine($"向kafka发送了数据: {deliveryReport.TopicPartitionOffset}");
                }
                catch (ProduceException<string, string> e)
                {
                    Console.WriteLine($"向kafka发送数据失败:{e.Message}");
                }
            }

            return Ok("OK");
        }

    }
}
